package com.atguigu.realtime.app.dws;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.bean.VideoChapterPlayVideoBean;
import com.atguigu.realtime.utils.*;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;

public class DwsVideoChapterPlayVideoWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
         */

        // TODO 3 读取dwd_traffic_appVideo_log中的数据
        String topicName = "dwd_traffic_appVideo_log";
        String groupId = "dws_video_chapter_playVideo_window";
        DataStreamSource<String> dataStream = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupId));

        // TODO 4 转换结构为JSONObject
        KeyedStream<JSONObject, String> keyedStream = dataStream.map(JSON::parseObject)
                .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts");
                    }
                }))
                .keyBy(new KeySelector<JSONObject, String>() {
                    @Override
                    public String getKey(JSONObject value) throws Exception {
                        return value.getJSONObject("appVideo").getString("video_id") + '-' + value.getJSONObject("common").getString("uid");
                    }
                });

//        keyedStream.print();

        // TODO 5 转换为javaBean，过滤数据
        SingleOutputStreamOperator<VideoChapterPlayVideoBean> processStream = keyedStream.process(new KeyedProcessFunction<String, JSONObject, VideoChapterPlayVideoBean>() {
            ValueState<Long> maxPositionState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<Long> maxPositionDescriptor = new ValueStateDescriptor<>("maxPosition", Long.class);
                // 给状态设置存活时间
                maxPositionDescriptor.enableTimeToLive(StateTtlConfig
                        // 设置TTL时间为1天
                        .newBuilder(Time.days(1L))
                        // 设置更新一天时间的条件为第一次创建和修改的时间
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        .build());
                maxPositionState = getRuntimeContext().getState(maxPositionDescriptor);
            }

            @Override
            public void processElement(JSONObject value, Context ctx, Collector<VideoChapterPlayVideoBean> out) throws Exception {
                Long maxPosition = maxPositionState.value();
                Long position_sec = value.getJSONObject("appVideo").getLong("position_sec");
                Long play_sec = value.getJSONObject("appVideo").getLong("play_sec");
                String video_id = value.getJSONObject("appVideo").getString("video_id");
                String uid = value.getJSONObject("common").getString("uid");
                VideoChapterPlayVideoBean videoChapterPlayVideoBean = new VideoChapterPlayVideoBean();
                videoChapterPlayVideoBean.setVideoId(video_id);
                videoChapterPlayVideoBean.setUid(uid);
                if (maxPosition == null) {
                    maxPosition = 0L;
                }
                if (play_sec < 30 || (play_sec == 30 && position_sec < maxPosition)) {
                    videoChapterPlayVideoBean.setVideoPlayCount(1L);
                    maxPositionState.update(position_sec);
                    videoChapterPlayVideoBean.setPlayDurSum(position_sec);
                }else{
                    videoChapterPlayVideoBean.setVideoPlayCount(0L);
                    videoChapterPlayVideoBean.setPlayDurSum(0L);
                }
                out.collect(videoChapterPlayVideoBean);
            }
        });

//        processStream.print("pro>>>>>>");
        // TODO 6 和dim层维度表进行关联

        SingleOutputStreamOperator<VideoChapterPlayVideoBean> VideoWithDimStream = processStream.map(new RichMapFunction<VideoChapterPlayVideoBean, VideoChapterPlayVideoBean>() {
            DruidDataSource dataSource = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                dataSource = DruidPhoenixDSUtil.getDataSource();
            }

            @Override
            public VideoChapterPlayVideoBean map(VideoChapterPlayVideoBean value) throws Exception {
                DruidPooledConnection connection = dataSource.getConnection();
                //关联视频表
                List<JSONObject> videoInfo = DimUtil.getDimInfo(connection, "DIM_VIDEO_INFO", value.getVideoId());
//                System.out.println("videoInfo:" + videoInfo.toString());
                if (videoInfo.size() > 0) {
                    JSONObject jsonObject = videoInfo.get(0);
                    String chapterId = jsonObject.getString("chapterId");
                    value.setChapterId(chapterId);
                }
                return value;
            }
        });

//        VideoWithDimStream.print("video>>>>>>>>>>>>");

        // TODO 7 分组开窗聚合
        SingleOutputStreamOperator<VideoChapterPlayVideoBean> reduceStream = VideoWithDimStream.keyBy(bean -> bean.getChapterId() + "/" + bean.getUid())
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)))
                .reduce(new ReduceFunction<VideoChapterPlayVideoBean>() {
                            @Override
                            public VideoChapterPlayVideoBean reduce(VideoChapterPlayVideoBean value1, VideoChapterPlayVideoBean value2) throws Exception {
                                value1.setVideoPlayCount(value1.getVideoPlayCount() + value2.getVideoPlayCount());
                                value1.setPlayDurSum(value1.getPlayDurSum() + value2.getPlayDurSum());
                                return value1;
                            }
                        },
                        new WindowFunction<VideoChapterPlayVideoBean, VideoChapterPlayVideoBean, String, TimeWindow>() {
                            @Override
                            public void apply(String s, TimeWindow window, Iterable<VideoChapterPlayVideoBean> input, Collector<VideoChapterPlayVideoBean> out) throws Exception {
                                VideoChapterPlayVideoBean next = input.iterator().next();
                                next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                                next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                                next.setTs(System.currentTimeMillis());
                                out.collect(next);
                            }
                        });

        reduceStream.print("reduce>>>>>>");

        // TODO 8 写入至clickhouse
        reduceStream.addSink(ClickHouseUtil.<VideoChapterPlayVideoBean>getClickHouseSinkFunc("insert into dws_video_chapter_playVideo_window values(?,?,?,?,?,?,?,?)"));

        env.execute();

    }
}
